package org.niit.service

import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.DStream
import org.niit.bean.AdClickData
import org.niit.common.TService
import org.niit.dao.BlackListDao

import java.text.SimpleDateFormat
import java.util.Date

/**
 * Date:2025/5/30
 * Author：Ys
 * Description:
 */
class BlackListService extends TService[DStream[AdClickData]]{


  //创建一个总的方法，在这个总的方法中去调用分支方法




 override def dataAnalysis(data:DStream[AdClickData]): Unit ={

    //调用各个分支方法

    //1.将从Kafka中获取的数据进行过滤，过滤掉不合法的数据
    val ds: DStream[((String, String, String), Int)] = filterKafkaData(data)

    //2.检查用户点击次数 满足条件拉黑
    checkUserCount(ds)

  }
  /*
     利用在Kafka中采集出来的流式数据对数据进行实时检索
     将已经在黑名单中的数据过滤掉，过滤掉后的数据存进每日用户点击广告次数表中
   */
  def filterKafkaData(data: DStream[AdClickData]): DStream[((String,String,String),Int)] = {
    // transform  转换RDD 顺带回顾下 rdd的基本操作
    data.transform(rdd=>{//rdd[AdClickData]

      //filterRdd 过滤掉在黑名单的数据
      val filterRdd = rdd.filter(line=>{
        //把不在黑名单中的数据进行返回 ---》 这里是操作数据库
        val blackListDao = new BlackListDao()
        !blackListDao.selectBlackUserById(line.user)
      })

      val redData = filterRdd.map(x=>{
        //获得用户ID 和 广告ID
        val user = x.user
        val ad = x.ad
        //时间格式转换  -- 数据中的时间戳 转换 --》 年-月-日
        //定义要转换的格式
        val sdf = new SimpleDateFormat("yyyy-MM-dd")
        val day = sdf.format(new Date(x.ts.toLong))

         //每一条数据当前用户在该时间内点击该广告的 1 次  数据最终是以key - value
         //   Key            Value
        (  ( day,user,ad ) , 1   )
      })
      /*
      (  (2025,小明,加多宝) ,1 )
        (  (2025,小明,加多宝) ,1 )
          (  (2025,小明,加多宝) ,1 )
            (  (2025,小明,加多宝) ,1 )
              (  (2025,小明,加多宝) ,1 )
                (  (2025,小明,加多宝) ,1 )
                                 ===> (  (2025,小明,加多宝) ,6 )
       */

      // 获得某用户在某天点击某广告的总次数
      val resData: RDD[((String, String, String), Int)] = redData.reduceByKey(_ + _)

      resData

    })

  }
  /*
   检查用户点击次数  如果不满足拉入黑名单的条件则存入每日用户点击广告次数表中，当存入表后满足拉入黑名单的条件则拉入黑名单中
   */
  def checkUserCount(ds: DStream[((String, String, String), Int)]): Unit = {
    //DStream 没有输出语句可以执行吗？ 不能  --》 输出有几种： print foreachRDD
    ds.foreachRDD(rdd=>{ // rdd==> DStream   集合 装了之前处理好的数据集合

      rdd.foreach{
        case ((day,user,ad),count) =>{
          //如果之前的统计点击次数大于20次，则拉入黑名单中
          val blackListDao = new BlackListDao()
          if(count >= 20){
            //拉入黑名单中  ---> 数据库的操作
            blackListDao.insertBlackList(user)
          }else{
            // 存入每日用户点击广告次数表中 ---> 数据库的操作
            blackListDao.insertUserAdCount(day,user,ad,count)
          }
          //检查(在数据库中查询)该改该用户点击该广告是否超过20次 ,如果超过就拉黑
          val bool: Boolean = blackListDao.checkUserByCount(day, user, ad, 20)
          if (bool){
            //拉入黑名单中  ---> 数据库的操作
            blackListDao.insertBlackList(user)
          }
        }
      }

    })

  }
}
